package org.example.supervise.supervisioncenter.holographicarchives

import org.apache.spark.broadcast.Broadcast
import org.apache.spark.sql.SparkSession
import org.example.client.DbClient
import org.example.common.{Logging, Sparking}
import org.example.constant.ApolloConst
import org.example.utils.Tools
import org.joda.time.DateTime
import org.joda.time.format.DateTimeFormat
import scalikejdbc.{NamedDB, SQL}

import java.sql.Timestamp
import java.util.Properties

/**
 * 车辆档案,推送到 zcov.yz_vehicle_info_archive
 */
object VehicleInfoArchive extends Sparking with Logging{
  def main(args: Array[String]): Unit = {
    val session = SparkSession.builder().config(conf)
      .config("hive.metastore.uris", ApolloConst.hiveMetastore)
      .enableHiveSupport()
      .getOrCreate()
    val todayStr: String = new DateTime().toString("yyyy-MM-dd")
    val yesterdayStr: String = DateTime.parse(todayStr, DateTimeFormat.forPattern("yyyy-MM-dd")).plusDays(-1).toString("yyyy-MM-dd")
    import session.implicits._
    val prop = new Properties()
    prop.put("user", ApolloConst.jgdMysqlUserName)
    prop.put("password", ApolloConst.jgdMysqlPassWord)
    prop.put("driver", ApolloConst.jgdMysqlDriver)
    //企业云同步到监管端的企业基本信息
    session.read.jdbc(ApolloConst.jgdMysqlURL, "zcov.basic_enterprise_info", prop).toDF().createOrReplaceTempView("base_into_enterprise_info")
    //企业云同步到监管端的驾驶人基本信息
    session.read.jdbc(ApolloConst.jgdMysqlURL, "zcov.basic_driver_info", prop).toDF().createOrReplaceTempView("base_transport_driver")
    //企业云同步到监管端的车辆基本信息
    session.read.jdbc(ApolloConst.jgdMysqlURL, "zcov.basic_vehicle_info", prop).createOrReplaceTempView("base_into_vehicle_info")
    //经营范围字典数据获取，写入到Map中
    val scopeCodeNameMap: Map[String, String] = session.sql("select business_code,business_scope_name from dim.business_scope_dic").map { row =>
      val business_code: String = row.getAs[String]("business_code")
      val business_scope_name: String = row.getAs[String]("business_scope_name")
      (business_code, business_scope_name)
    }.collect().toMap
    val scopeCodeNameBroadcast: Broadcast[Map[String, String]] = session.sparkContext.broadcast(scopeCodeNameMap)
    //同步车辆档案到监管端mysql：zcov.yz_vehicle_info_archive
    DbClient.init("jdgMysql", ApolloConst.jgdMysqlDriver, ApolloConst.jgdMysqlURL, ApolloConst.jgdMysqlUserName, ApolloConst.jgdMysqlPassWord)
    DbClient.usingDB("jdgMysql") { db: NamedDB =>
      val sqlStr =
        s"""
           |truncate table zcov.yz_vehicle_info_archive
           |""".stripMargin
      db autoCommit { implicit session =>
        SQL(sqlStr).update().apply()
      }
    }
    val frame3 = session.sql(
      """
        |select
        |   t1.vehicle_number plate_number, -- 车牌号
        |   case when t1.vehicle_plate_color ='蓝色' then '1'
        |        when t1.vehicle_plate_color ='黄色' then '2'
        |        when t1.vehicle_plate_color ='黑色' then '3'
        |        when t1.vehicle_plate_color ='白色' then '4'
        |        when t1.vehicle_plate_color ='绿色' then '5'
        |        when t1.vehicle_plate_color ='渐变绿色' then '6'
        |        else '9' end plate_color, -- 车牌颜色 1 蓝 2 黄 3 黑 4 白 5 绿 6 渐变绿色 9 其他
        |   t3.social_credit_code, -- 业户统一社会信用代码
        |   t2.business_owner_name, -- 业户名称/所属企业
        |   t1.operation_state operate_state, -- 营运状态     在车辆表中营运状态全为空
        |
        |   t4.road_transportation_certificate_classification_number license_number, -- 道路运输证号
        |   cast(t4.effective_date as string) issuing_date, -- 有效期始/发证日期
        |   cast(t4.expire_date as string) invalid_date, -- 有效期至
        |   t4.license_issuing_authority issuing_authority, -- 发证机关
        |   t4.business_scope, -- 经营范围
        |   t4.road_transportation_certificate_classification_word transport_field, -- 道路运输证字
        |
        |   t1.vehicle_address address, -- 地址
        |   t1.vehicle_type, -- 车辆类型
        |   if(t1.positioning_system_installation_status = '1','是','否') positioning_system_installation_status ,  -- 定位系统安装状态(+)
        |   t1.positioning_pattern, -- 定位模式（+）
        |   t1.positioning_platform_operator, -- 定位平台运营商+
        |   t1.equipment_mode,  -- 卫星定位设备型号+
        |   t1.curb_total_mass, -- 整备总质量KG+
        |   case when t1.axle_number ='一轴' then '1'
        |        when t1.axle_number ='两轴' then '2'
        |        when t1.axle_number ='三轴' then '3'
        |        when t1.axle_number ='四轴' then '4'
        |        when t1.axle_number ='五轴' then '5'
        |        when t1.axle_number ='六轴' then '6'
        |        when t1.axle_number ='七轴' then '7'
        |        when t1.axle_number ='八轴' then '8'
        |        else '0' end axle_number,-- 车轴数+
        |   t1.buy_car_date,  -- 购车日期+
        |   t1.product_date,  -- 出厂日期+
        |   t1.fuel_type, -- 燃料类型+
        |   t1.vehicle_identification_number, -- 车架号+
        |   t1.engine_number, -- 发动机号+
        |   t1.engine_model,  -- 发动机型号+
        |   t1.vehicle_color, -- 车身颜色+
        |   t1.vecent_type, -- 实体类别+
        |   t1.model, -- 型号+
        |   t1.factory_plate, -- 厂牌+
        |   t1.own_organization, -- 所属机构+
        |   t1.file_create_date,  -- 建档日期+
        |   t1.total_mass,  -- 总质量+
        |   t1.register_date, -- 注册日期+
        |   t1.positioning_platform_name, -- 车辆接入卫星定位平台名称+
        |   t1.positioning_platform_code, -- 车辆接入卫星定位平台代码+
        |   t1.whether_install, -- 卫星定位装置是否安装+
        |   t1.frist_licence_date,  -- 初次领证日期+
        |   t1.objid,  -- 图片id+
        |   t4.licence_type,  -- 证照类型+
        |   t4.vehicle_archives_number,  -- 车辆档案号
        |   case when t1.ratified_seats is null or t1.ratified_seats = '' then 0
        |        when t1.ratified_seats <= 0 then abs(t1.ratified_seats)
        |        when t1.ratified_seats >0 then t1.ratified_seats
        |        else 0 end seating, -- 座位数
        |   case when t1.approved_load_quality is null or t1.approved_load_quality = '' then 0
        |        when t1.approved_load_quality <= 0 then abs(t1.approved_load_quality)
        |        when t1.approved_load_quality >0 then t1.approved_load_quality
        |        else 0 end tonnage, -- 吨位
        |   case when t1.approved_load_quality is null or t1.approved_load_quality = '' then 0
        |        when t1.approved_load_quality <= 0 then abs(t1.approved_load_quality)
        |        when t1.approved_load_quality >0 then t1.approved_load_quality
        |        else 0 end approved_load_quality, -- 核定载质量（+）
        |   case when t1.vehicle_length is null or t1.vehicle_length = '' then 0
        |        when t1.vehicle_length <= 0 then abs(t1.vehicle_length)
        |        when t1.vehicle_length > 0 then t1.vehicle_length
        |        else 0 end length, -- 长（mm）
        |   case when t1.vehicle_wide is null or t1.vehicle_wide = ''  then 0
        |        when t1.vehicle_wide <= 0 then abs(t1.vehicle_wide)
        |        when t1.vehicle_wide > 0 then t1.vehicle_wide
        |        else 0 end width, -- 宽（mm）
        |   case when t1.vehicle_high is null or t1.vehicle_high = ''  then 0
        |        when t1.vehicle_high <= 0 then abs(t1.vehicle_high)
        |        when t1.vehicle_high > 0 then t1.vehicle_high
        |        else 0 end height, -- 高（mm）
        |   t5.imageaddr vehicle_photo_url -- 车辆照片url
        |from
        |  (
        |  SELECT * FROM
        |     (SELECT
        |        *,
        |        ROW_NUMBER() OVER(PARTITION BY vehicle_number,vehicle_plate_color ORDER BY register_date DESC) rk
        |      FROM dwd.dwd_yz_vehicle_info
        |      where TRIM(vehicle_number) !=''
        |      and TRIM(vehicle_plate_color) !=''
        |      -- and vehicle_wide > 0
        |      -- and vehicle_high > 0
        |      -- and vehicle_length > 0
        |      -- and approved_load_quality > 0
        |      -- and ratified_seats > 0
        |      and TRIM(business_owner_id) != ''
        |      and business_owner_id is not null
        |      and trim(vehicle_id) != ''
        |      and vehicle_id is not null )t
        |   where t.rk = 1
        |  )t1
        |left join
        |  (
        |   SELECT * from
        |      (SELECT
        |          business_owner_id,
        |          business_owner_name,
        |          ROW_NUMBER() OVER(PARTITION BY business_owner_id ORDER BY create_date DESC) rk
        |       from dwd.dwd_yz_company_info
        |       where TRIM(business_owner_id) != ''
        |       and business_owner_id is not NULL) t
        |   where t.rk =1
        |  )t2
        |  on  t1.business_owner_id = t2.business_owner_id
        |left join
        |  (
        |   SELECT
        |      business_owner_id,
        |      case WHEN trim(business_registration_number) != '' and business_registration_number is not NULL then trim(business_registration_number)
        |      WHEN (trim(business_registration_number) = '' or business_registration_number is NULL ) and (trim(organization_code_card) != '' and organization_code_card is not NULL) then trim(organization_code_card)
        |      WHEN (trim(business_registration_number) = '' or business_registration_number is NULL)  and (trim(organization_code_card) = '' or organization_code_card is NULL ) and (trim(organization_code) != '' and organization_code is not NULL) then trim(organization_code)
        |      END social_credit_code
        |   from
        |      dwd.dwd_yz_business_info
        |   where trim(business_owner_id) != ''
        |   and business_owner_id is not null
        |   and (LENGTH(TRIM(organization_code)) >1
        |      or LENGTH(TRIM(organization_code_card)) >1
        |      or LENGTH(TRIM(business_registration_number)) >1)
        |  )t3
        |  on t1.business_owner_id = t3.business_owner_id
        |left join
        |  (
        |   SELECT   *
        |   from dwd.dwd_yz_road_transport_certificate_info
        |   where TRIM(road_transportation_certificate_classification_number) !=''
        |   AND road_transportation_certificate_classification_number is not NULL
        |   AND TRIM(vehicle_id) !=''
        |   AND vehicle_id is not null
        |  )t4
        |on t1.vehicle_id = t4.vehicle_id
        |left join
        |  (
        |   SELECT   objid,imageaddr
        |   from dwd.dwd_vehicle_imageaddr_info
        |  )t5
        |on t1.objid = t5.objid
        |""".stripMargin).toDF()
    frame3.createOrReplaceTempView("yz_vehicle_table")

    session.sql(
      """
        |select id, vehicle_code, plate_num plate_number, plate_color, vehicle_custom_num, into_time, into_status,
        |business_scope, business_scope_detail, indentifier, operating_certificate_no, class_line, use_nature,
        |t1.enterprise_code, driving_permit_no, status, vehicle_brand, insert_time, update_time, platform_code,
        |is_deleted, control_type, is_sync, front_photo, back_photo, drive_license_photo, vehicle_register_cert,
        |compulsory_liability_insurance, other_photo, engine_no, vehicle_id_num, vehicle_type, owner, register_date,
        |approve_seats, business_state, record_code, record_card_url, audit_state, record_date, invalid_date, business_model
        |,t2.social_credit_code,t2.enterprise_name
        | from
        |(SELECT id, vehicle_code, plate_num, plate_color, vehicle_custom_num, into_time, into_status,
        |business_scope, business_scope_detail, indentifier, operating_certificate_no, class_line, use_nature,
        |enterprise_code, driving_permit_no, status, vehicle_brand, gmt_create as insert_time, gmt_modified as update_time, platform_code,
        |deleted as is_deleted, control_type, is_sync, front_photo, back_photo, drive_license_photo, vehicle_register_cert,
        |compulsory_liability_insurance, other_photo, engine_no, vehicle_id_num, vehicle_type, owner, register_date,
        |approve_seats, business_state, record_code, record_card_url, audit_state, record_date, invalid_date, business_model
        |from base_into_vehicle_info
        |) t1
        |left join
        |(select enterprise_code,social_credit_code,enterprise_name from base_into_enterprise_info ) t2
        |on t1.enterprise_code=t2.enterprise_code
        |""".stripMargin).createOrReplaceTempView("qiyeyun_vehicle_table")


    val frame = session.sql(
      """
        |select
        |plate_number,
        |plate_color,
        |coalesce(social_credit_code_arr[0],social_credit_code_arr[1]) social_credit_code,
        |coalesce(business_owner_name_arr[0],business_owner_name_arr[1]) business_owner_name,
        |coalesce(enter_flag_arr[0],enter_flag_arr[1]) enter_flag,
        |coalesce(insert_time_arr[0],insert_time_arr[1]) insert_time,
        |coalesce(platform_code_arr[0],platform_code_arr[1]) platform_code,
        |coalesce(use_nature_arr[0],use_nature_arr[1]) use_nature,
        |coalesce(control_type_arr[0],control_type_arr[1]) control_type,
        |coalesce(operate_state_arr[0],operate_state_arr[1]) operate_state,
        |coalesce(license_number_arr[0],license_number_arr[1]) license_number,
        |coalesce(issuing_date_arr[0],issuing_date_arr[1]) issuing_date,
        |coalesce(invalid_date_arr[0],invalid_date_arr[1]) invalid_date,
        |coalesce(issuing_authority_arr[0],issuing_authority_arr[1]) issuing_authority,
        |coalesce(business_scope_arr[0],business_scope_arr[1]) business_scope,
        |coalesce(transport_field_arr[0],transport_field_arr[1]) transport_field,
        |coalesce(address_arr[0],address_arr[1]) address,
        |coalesce(vehicle_type_arr[0],vehicle_type_arr[1]) vehicle_type,
        |coalesce(seating_arr[0],seating_arr[1]) seating,
        |coalesce(tonnage_arr[0],tonnage_arr[1]) tonnage,
        |coalesce(length_arr[0],length_arr[1]) length,
        |coalesce(width_arr[0],width_arr[1]) width,
        |coalesce(height_arr[0],height_arr[1]) height,
        |coalesce(vehicle_photo_url_arr[0],vehicle_photo_url_arr[1]) vehicle_photo_url,
        |coalesce(approved_load_quality_arr[0],approved_load_quality_arr[1]) approved_load_quality,
        |coalesce(positioning_system_installation_status_arr[0],positioning_system_installation_status_arr[1]) positioning_system_installation_status,
        |coalesce(positioning_pattern_arr[0],positioning_pattern_arr[1]) positioning_pattern,
        |coalesce(positioning_platform_operator_arr[0],positioning_platform_operator_arr[1]) positioning_platform_operator,
        |coalesce(equipment_mode_arr[0],equipment_mode_arr[1]) equipment_mode,
        |coalesce(curb_total_mass_arr[0],curb_total_mass_arr[1]) curb_total_mass,
        |coalesce(axle_number_arr[0],axle_number_arr[1]) axle_number,
        |coalesce(buy_car_date_arr[0],buy_car_date_arr[1]) buy_car_date,
        |coalesce(product_date_arr[0],product_date_arr[1]) product_date,
        |coalesce(fuel_type_arr[0],fuel_type_arr[1]) fuel_type,
        |coalesce(vehicle_identification_number_arr[0],vehicle_identification_number_arr[1]) vehicle_identification_number,
        |coalesce(engine_number_arr[0],engine_number_arr[1]) engine_number,
        |coalesce(engine_model_arr[0],engine_model_arr[1]) engine_model,
        |coalesce(vehicle_color_arr[0],vehicle_color_arr[1]) vehicle_color,
        |coalesce(vecent_type_arr[0],vecent_type_arr[1]) vecent_type,
        |coalesce(model_arr[0],model_arr[1]) model,
        |coalesce(factory_plate_arr[0],factory_plate_arr[1]) factory_plate,
        |coalesce(own_organization_arr[0],own_organization_arr[1]) own_organization,
        |coalesce(file_create_date_arr[0],file_create_date_arr[1])   file_create_date,
        |coalesce(total_mass_arr[0],total_mass_arr[0]) total_mass,
        |coalesce(register_date_arr[0],register_date_arr[0])  register_date,
        |coalesce(positioning_platform_name_arr[0],positioning_platform_name_arr[1]) positioning_platform_name,
        |coalesce(positioning_platform_code_arr[0],positioning_platform_code_arr[1])  positioning_platform_code,
        |coalesce(whether_install_arr[0],whether_install_arr[1]) whether_install,
        |coalesce(frist_licence_date_arr[0],frist_licence_date_arr[1])  frist_licence_date,
        |coalesce(licence_type_arr[0],licence_type_arr[1]) licence_type,
        |coalesce(vehicle_archives_number_arr[0],vehicle_archives_number_arr[1])   vehicle_archives_number,
        |coalesce(objid_arr[0],objid_arr[1]) objid,
        |data_resource
        |from
        |(select
        |plate_number,
        |plate_color,
        |COLLECT_list(social_credit_code) social_credit_code_arr,
        |COLLECT_list(business_owner_name) business_owner_name_arr,
        |COLLECT_list(enter_flag) enter_flag_arr,
        |COLLECT_list(insert_time) insert_time_arr,
        |COLLECT_list(platform_code) platform_code_arr,
        |COLLECT_list(use_nature) use_nature_arr,
        |COLLECT_list(control_type) control_type_arr,
        |COLLECT_list(operate_state) operate_state_arr,
        |COLLECT_list(license_number) license_number_arr,
        |COLLECT_list(issuing_date) issuing_date_arr,
        |COLLECT_list(invalid_date) invalid_date_arr,
        |COLLECT_list(issuing_authority) issuing_authority_arr,
        |COLLECT_list(business_scope) business_scope_arr,
        |COLLECT_list(transport_field) transport_field_arr,
        |COLLECT_list(address) address_arr,
        |COLLECT_list(vehicle_type) vehicle_type_arr,
        |COLLECT_list(seating) seating_arr,
        |COLLECT_list(tonnage) tonnage_arr,
        |COLLECT_list(length) length_arr,
        |COLLECT_list(width) width_arr,
        |COLLECT_list(height) height_arr,
        |COLLECT_list(vehicle_photo_url) vehicle_photo_url_arr,
        |COLLECT_list(approved_load_quality) approved_load_quality_arr,
        |COLLECT_list(positioning_system_installation_status) positioning_system_installation_status_arr,
        |COLLECT_list(positioning_pattern) positioning_pattern_arr,
        |COLLECT_list(positioning_platform_operator) positioning_platform_operator_arr,
        |COLLECT_list(equipment_mode) equipment_mode_arr,
        |COLLECT_list(curb_total_mass) curb_total_mass_arr,
        |COLLECT_list(axle_number) axle_number_arr,
        |COLLECT_list(buy_car_date) buy_car_date_arr,
        |COLLECT_list(product_date) product_date_arr,
        |COLLECT_list(fuel_type) fuel_type_arr,
        |COLLECT_list(vehicle_identification_number) vehicle_identification_number_arr,
        |COLLECT_list(engine_number) engine_number_arr,
        |COLLECT_list(engine_model) engine_model_arr,
        |COLLECT_list(vehicle_color) vehicle_color_arr,
        |COLLECT_list(vecent_type) vecent_type_arr,
        |COLLECT_list(model) model_arr,
        |COLLECT_list(factory_plate) factory_plate_arr,
        |COLLECT_list(own_organization) own_organization_arr,
        |COLLECT_list(file_create_date)   file_create_date_arr,
        |COLLECT_list(total_mass) total_mass_arr,
        |COLLECT_list(register_date)  register_date_arr,
        |COLLECT_list(positioning_platform_name) positioning_platform_name_arr,
        |COLLECT_list(positioning_platform_code)  positioning_platform_code_arr,
        |COLLECT_list(whether_install) whether_install_arr,
        |COLLECT_list(frist_licence_date)  frist_licence_date_arr,
        |COLLECT_list(licence_type) licence_type_arr,
        |COLLECT_list(vehicle_archives_number)   vehicle_archives_number_arr,
        |COLLECT_list(objid) objid_arr,
        |sum(data_resource) data_resource
        |from
        |(select
        |   plate_number, -- 车牌号
        |   cast(plate_color as int) plate_color, -- 车牌颜色 1 蓝 2 黄 3 黑 4 白 5 绿 9 其他
        |   social_credit_code, -- 业户统一社会信用代码
        |   business_owner_name, -- 业户名称/所属企业
        |   null enter_flag, -- 企业云-是否录入(云平台)：1是0否
        |   null insert_time, -- 企业云-录入时间
        |   null platform_code, -- 企业云-录入平台编码
        |   null use_nature, -- 企业云-使用性质
        |   null control_type, -- 企业云-车辆管控类型
        |   operate_state, -- 营运状态     在车辆表中营运状态全为空
        |   license_number, -- 道路运输证号
        |   issuing_date, -- 有效期始/发证日期
        |   invalid_date, -- 有效期至
        |   issuing_authority, -- 发证机关
        |   business_scope, -- 经营范围
        |   transport_field, -- 道路运输证字
        |   address, -- 地址
        |   vehicle_type, -- 车辆类型
        |   seating , -- 座位数
        |   tonnage , -- 吨位
        |   length , -- 长（mm）
        |   width , -- 宽（mm）
        |   height , -- 高（mm）
        |   vehicle_photo_url, -- 车辆照片url
        |
        |   approved_load_quality, -- 核定载质量（+）
        |   positioning_system_installation_status ,  -- 定位系统安装状态(+)
        |   positioning_pattern, -- 定位模式（+）
        |   positioning_platform_operator, -- 定位平台运营商+
        |   equipment_mode,  -- 卫星定位设备型号+
        |   curb_total_mass, -- 整备总质量KG+
        |   axle_number, -- 车轴数+
        |   buy_car_date,  -- 购车日期+
        |   product_date,  -- 出厂日期+
        |   fuel_type, -- 燃料类型+
        |   vehicle_identification_number, -- 车架号+
        |   engine_number, -- 发动机号+
        |   engine_model,  -- 发动机型号+
        |   vehicle_color, -- 车身颜色+
        |   vecent_type, -- 实体类别+
        |   model, -- 型号+
        |   factory_plate, -- 厂牌+
        |   own_organization, -- 所属机构+
        |   file_create_date,  -- 建档日期+
        |   total_mass,  -- 总质量+
        |   register_date, -- 注册日期+
        |   positioning_platform_name, -- 车辆接入卫星定位平台名称+
        |   positioning_platform_code, -- 车辆接入卫星定位平台代码+
        |   whether_install, -- 卫星定位装置是否安装+
        |   frist_licence_date,  -- 初次领证日期+
        |   licence_type,  -- 证照类型+
        |   vehicle_archives_number,  -- 车辆档案号
        |   objid,  -- 图片id
        |   1 data_resource
        |   from yz_vehicle_table
        |
        |   union
        |
        |   select
        |   plate_number, -- 车牌号
        |   cast(plate_color as int) plate_color, -- 车牌颜色 1 蓝 2 黄 3 黑 4 白 5 绿 9 其他
        |   social_credit_code, -- 业户统一社会信用代码
        |   enterprise_name business_owner_name, -- 业户名称/所属企业
        |   if(plate_number is null , 0, 1) enter_flag, -- 企业云-是否录入(云平台)：1是0否
        |   insert_time, -- 企业云-录入时间
        |   platform_code, -- 企业云-录入平台编码
        |   use_nature, -- 企业云-使用性质
        |   control_type, -- 企业云-车辆管控类型
        |   null operate_state, -- 营运状态     在车辆表中营运状态全为空
        |   null license_number, -- 道路运输证号
        |   null issuing_date, -- 有效期始/发证日期
        |   null invalid_date, -- 有效期至
        |   null issuing_authority, -- 发证机关
        |   null business_scope, -- 经营范围
        |   null transport_field, -- 道路运输证字
        |   null address, -- 地址
        |   null vehicle_type, -- 车辆类型
        |   null seating , -- 座位数
        |   null tonnage , -- 吨位
        |   null length , -- 长（mm）
        |   null width , -- 宽（mm）
        |   null height , -- 高（mm）
        |   null vehicle_photo_url, -- 车辆照片url
        |
        |   null approved_load_quality, -- 核定载质量（+）
        |   null positioning_system_installation_status ,  -- 定位系统安装状态(+)
        |   null positioning_pattern, -- 定位模式（+）
        |   null positioning_platform_operator, -- 定位平台运营商+
        |   null equipment_mode,  -- 卫星定位设备型号+
        |   null curb_total_mass, -- 整备总质量KG+
        |   null axle_number, -- 车轴数+
        |   null buy_car_date,  -- 购车日期+
        |   null product_date,  -- 出厂日期+
        |   null fuel_type, -- 燃料类型+
        |   null vehicle_identification_number, -- 车架号+
        |   null engine_number, -- 发动机号+
        |   null engine_model,  -- 发动机型号+
        |   null vehicle_color, -- 车身颜色+
        |   null vecent_type, -- 实体类别+
        |   null model, -- 型号+
        |   null factory_plate, -- 厂牌+
        |   null own_organization, -- 所属机构+
        |   null file_create_date,  -- 建档日期+
        |   null total_mass,  -- 总质量+
        |   null register_date, -- 注册日期+
        |   null positioning_platform_name, -- 车辆接入卫星定位平台名称+
        |   null positioning_platform_code, -- 车辆接入卫星定位平台代码+
        |   null whether_install, -- 卫星定位装置是否安装+
        |   null frist_licence_date,  -- 初次领证日期+
        |   null licence_type,  -- 证照类型+
        |   null vehicle_archives_number,  -- 车辆档案号
        |   null objid,  -- 图片id
        |   2 data_resource --数据来源
        |  from qiyeyun_vehicle_table) t1
        |  group by plate_number,plate_color ) t
        |""".stripMargin).toDF()

    frame.foreachPartition { iter =>
      DbClient.init("jdgMysql", ApolloConst.jgdMysqlDriver, ApolloConst.jgdMysqlURL, ApolloConst.jgdMysqlUserName, ApolloConst.jgdMysqlPassWord)
      iter.foreach { row =>
        val plate_number: String = row.getAs[String]("plate_number")
        val plate_color: Int = row.getAs[Int]("plate_color")
        val social_credit_code: String = row.getAs[String]("social_credit_code")
        val business_owner_name: String = row.getAs[String]("business_owner_name")
        val enter_flag: Int = row.getAs[Int]("enter_flag")
        val insert_time_timestamp = row.getAs[Timestamp]("insert_time")
        val insert_timeOption: Option[String] = Tools.parseString(insert_time_timestamp)
        val insert_time = insert_timeOption match {
          case Some(i) => i
          case None => null
        }
        val platform_code: String = row.getAs[String]("platform_code")
        var use_nature: String = row.getAs[String]("use_nature")
        var control_type: Int = row.getAs[Int]("control_type")
        val operate_state: String = row.getAs[String]("operate_state")
        val license_number: String = row.getAs[String]("license_number")
        val issuing_date_string: String = row.getAs[String]("issuing_date")
        val issuing_date = if (issuing_date_string != null && Tools.isDate(issuing_date_string.trim, "yyyy-MM-dd")) {
          issuing_date_string
        } else null

        //val issuing_date: String = if(issuingDate != null) new DateTime(issuingDate.getTime).toString("yyyy-MM-dd") else null
        val invalid_date_string: String = row.getAs[String]("invalid_date")
        val invalid_date = if (invalid_date_string != null && Tools.isDate(invalid_date_string.trim, "yyyy-MM-dd")) {
          invalid_date_string
        } else null

        //val invalid_date: String = if (invalidDate != null) new DateTime(invalidDate.getTime).toString("yyyy-MM-dd") else null
        val issuing_authority: String = row.getAs[String]("issuing_authority")
        val business_scope: String = row.getAs[String]("business_scope")
        if (use_nature == null && business_scope != null) {

          if (business_scope.contains("班车客运")) {
            use_nature = "13"
            control_type = 15
          } else if (business_scope.contains("包车客运") || business_scope.contains("旅游客运")) {
            use_nature = "16"
            control_type = -1
          }
          else if (business_scope.contains("危险货物")) {
            use_nature = "11"
            control_type = 18
          }
          else if (business_scope.contains("普通货物") || business_scope.contains("普通货运") || business_scope.contains("货物专用运输") || business_scope.contains("大型物件运输")) {
            use_nature = "19"
            control_type = 21
          }
          else if (business_scope.contains("A1") || business_scope.contains("A2") || business_scope.contains("A3")
            || business_scope.contains("B1") || business_scope.contains("B2") || business_scope.contains("B3")
            || business_scope.contains("c1") || business_scope.contains("c2") || business_scope.contains("E")) {
            //驾校类使用性质（企业端暂未给出使用性质）
          }
          else if (business_scope.contains("城市公共汽车")) {
            //城市公共汽车（企业端暂未给出使用性质）
          }
          else if (business_scope.contains("巡游出租")) {
            //出租车（企业端暂未给出使用性质）
          }
          else if (business_scope.contains("网络预约出租")) {
            //网约车（企业端暂未给出使用性质）
          }
        }


        val transport_field: String = row.getAs[String]("transport_field")
        val address: String = row.getAs[String]("address")
        val vehicle_type: String = row.getAs[String]("vehicle_type")
        var seating_string = row.getAs[String]("seating")
        var seatingOption: Option[Int] = Tools.parseInt(seating_string)
        val seating = seatingOption match {
          case Some(i) => i
          case None => 0
        }
        val tonnage_string = row.getAs[String]("tonnage")
        val tonnageOption: Option[Int] = Tools.parseInt(tonnage_string)
        val tonnage = tonnageOption match {
          case Some(i) => i
          case None => 0
        }
        val length_string = row.getAs[String]("length")
        val lengthOption: Option[Int] = Tools.parseInt(length_string)
        val length = lengthOption match {
          case Some(i) => i
          case None => 0
        }
        val width_string = row.getAs[String]("width")
        val widthOption: Option[Int] = Tools.parseInt(width_string)
        val width = widthOption match {
          case Some(i) => i
          case None => 0
        }
        val height_string = row.getAs[String]("height")
        val heightOption: Option[Int] = Tools.parseInt(height_string)
        val height = heightOption match {
          case Some(i) => i
          case None => 0
        }
        val vehicle_photo_url: String = row.getAs[String]("vehicle_photo_url")

        val approved_load_quality_string = row.getAs[String]("approved_load_quality")
        val approved_load_qualityOption: Option[Int] = Tools.parseInt(approved_load_quality_string)
        val approved_load_quality = approved_load_qualityOption match {
          case Some(i) => i
          case None => 0
        }
        val positioning_system_installation_status: String = row.getAs[String]("positioning_system_installation_status")
        val positioning_pattern: String = row.getAs[String]("positioning_pattern")
        val positioning_platform_operator: String = row.getAs[String]("positioning_platform_operator")
        val equipment_mode: String = row.getAs[String]("equipment_mode")
        val curb_total_mass_string: String = row.getAs[String]("curb_total_mass")
        val curb_total_mass = if (curb_total_mass_string != "" && curb_total_mass_string != null) {
          curb_total_mass_string.toDouble.toInt
        } else {
          0
        }
        val axle_number_string: String = row.getAs[String]("axle_number")
        val axle_number = if (axle_number_string != "" && axle_number_string != null) {
          axle_number_string.toDouble.toInt
        } else {
          0
        }

        val buy_car_date_string: String = row.getAs[String]("buy_car_date")
        val buy_car_date = if (buy_car_date_string != null && Tools.isDate(buy_car_date_string.trim, "yyyy-MM-dd")) {
          buy_car_date_string
        } else null

        val product_date_string: String = row.getAs[String]("product_date")
        val product_date = if (product_date_string != null && Tools.isDate(product_date_string.trim, "yyyy-MM-dd")) {
          product_date_string
        } else null

        val fuel_type: String = row.getAs[String]("fuel_type")
        val vehicle_identification_number: String = row.getAs[String]("vehicle_identification_number")
        val engine_number: String = row.getAs[String]("engine_number")
        val engine_model: String = row.getAs[String]("engine_model")
        val vehicle_color: String = row.getAs[String]("vehicle_color")
        val vecent_type: String = row.getAs[String]("vecent_type")
        val model: String = row.getAs[String]("model")
        val factory_plate: String = row.getAs[String]("factory_plate")
        val own_organization: String = row.getAs[String]("own_organization")
        val file_create_date_string: String = row.getAs[String]("file_create_date")
        val file_create_date = if (file_create_date_string != null && Tools.isDate(file_create_date_string.trim, "yyyy-MM-dd")) {
          file_create_date_string
        } else null

        val total_mass_string: String = row.getAs[String]("total_mass")
        val total_mass: Int = if (total_mass_string != "" && total_mass_string != null) {
          total_mass_string.toDouble.toInt
        } else {
          0
        }

        val register_date_string: String = row.getAs[String]("register_date")
        val register_date = if (register_date_string != null && Tools.isDate(register_date_string.trim, "yyyy-MM-dd")) {
          register_date_string
        } else null

        val positioning_platform_name: String = row.getAs[String]("positioning_platform_name")
        val positioning_platform_code: String = row.getAs[String]("positioning_platform_code")
        val whether_install: String = row.getAs[String]("whether_install")
        val frist_licence_date_string: String = row.getAs[String]("frist_licence_date")
        val frist_licence_date = if (frist_licence_date_string != null && Tools.isDate(frist_licence_date_string.trim, "yyyy-MM-dd")) {
          frist_licence_date_string
        } else null

        val licence_type: String = row.getAs[String]("licence_type")
        val vehicle_archives_number: String = row.getAs[String]("vehicle_archives_number")
        val objid: String = row.getAs[String]("objid")
        val data_resource: Int = row.getAs[Long]("data_resource").toInt

        val fieldList = List(plate_number,
          plate_color,
          social_credit_code,
          business_owner_name,
          enter_flag,
          insert_time,
          platform_code,
          use_nature,
          control_type,
          operate_state,
          license_number,
          issuing_date,
          invalid_date,
          issuing_authority,
          business_scope,
          transport_field,
          address,
          vehicle_type,
          seating,
          tonnage,
          length,
          width,
          height,
          vehicle_photo_url,
          approved_load_quality,
          positioning_system_installation_status,
          positioning_pattern,
          positioning_platform_operator,
          equipment_mode,
          curb_total_mass,
          axle_number,
          buy_car_date,
          product_date,
          fuel_type,
          vehicle_identification_number,
          engine_number,
          engine_model,
          vehicle_color,
          vecent_type,
          model,
          factory_plate,
          own_organization,
          file_create_date,
          total_mass,
          register_date,
          positioning_platform_name,
          positioning_platform_code,
          whether_install,
          frist_licence_date,
          licence_type,
          vehicle_archives_number,
          objid,
          data_resource
        )
        DbClient.usingDB("jdgMysql") { db: NamedDB =>
          val sqlStr =
            s"""
               |replace into zcov.yz_vehicle_info_archive(
               |plate_number,
               |plate_color,
               |social_credit_code,
               |business_owner_name,
               |enter_flag,
               |insert_time,
               |platform_code,
               |use_nature,
               |control_type,
               |operate_state,
               |license_number,
               |issuing_date,
               |invalid_date,
               |issuing_authority,
               |business_scope,
               |transport_field,
               |address,
               |vehicle_type,
               |seating,
               |tonnage,
               |length,
               |width,
               |height,
               |vehicle_photo_url,
               |approved_load_quality,
               |positioning_system_installation_status,
               |positioning_pattern,
               |positioning_platform_operator,
               |equipment_mode,
               |curb_total_mass,
               |axle_number,
               |buy_car_date,
               |product_date,
               |fuel_type,
               |vehicle_identification_number,
               |engine_number,
               |engine_model,
               |vehicle_color,
               |vecent_type,
               |model,
               |factory_plate,
               |own_organization,
               |file_create_date,
               |total_mass,
               |register_date,
               |positioning_platform_name,
               |positioning_platform_code,
               |whether_install,
               |frist_licence_date,
               |licence_type,
               |vehicle_archives_number,
               |objid,
               |data_resource) values(?,?,?,?,?, ?,?,?,?,?, ?,?,?,?,?, ?,?,?,?,?, ?,?,?,?,?, ?,?,?,?,?, ?,?,?,?,?, ?,?,?,?,?, ?,?,?,?,?, ?,?,?,?,?, ?,?,?)
               |""".stripMargin
          try {
            db autoCommit { implicit session =>
              SQL(sqlStr).bind(fieldList: _*).update().apply()
            }
          } catch {
            case _: Exception => error("报错数据：" + fieldList.mkString(","))
          }
        }
      }
    }
  }
}
